package com.henu.main.executors;

import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.henu.main.util.MetricsUtil;

import java.util.HashMap;
import java.util.Map;

public class KafkaLagExecutor implements Runnable {

    @Override
    public void run() {
        String html = HttpUtil.get("http://qh-prod-bd-realtime-2-150:8080/api/status/StreamingLighting/consumersSummary");
        JSONObject jsonObject = JSONUtil.parseObj(html);
        JSONArray consumers = (JSONArray) jsonObject.get("consumers");
        for (Object consumer : consumers) {
            JSONObject consumerObj = JSONUtil.parseObj(consumer.toString());
            String consumerName = consumerObj.get("name").toString();
            String lags = consumerObj.get("lags").toString();
            JSONObject lagsObj = JSONUtil.parseObj(lags);
            for (Map.Entry<String, Object> lag : lagsObj) {
                String topic = lag.getKey();
                Double value = Double.parseDouble(lag.getValue() + "");
                HashMap<String, Object> metric = new HashMap<>();
                metric.put("metricName", "KafkaLag");
                metric.put("consumer", consumerName);
                metric.put("topic", topic);
                metric.put("value", value);
                try {
                    MetricsUtil.queue.put(metric);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

    }

}
